package onehour.pulsar.examples;

import org.apache.pulsar.client.api.*;

import java.util.Random;
import java.util.concurrent.TimeUnit;

public class OrderConsumer {

    public static void main(String[] args) throws Exception {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();


        Consumer consumer = client.newConsumer(Schema.STRING)
                .topic("my-order")
                .subscriptionName("my-subscription")
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true) // 开启重试(延时)队列
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        .retryLetterTopic("retry-order") // 重试队列名称
                        .maxRedeliverCount(1)  // 超过最大重试次数，将消息放入死信队列
                        .deadLetterTopic("dead-order") //死信队列名称
                        .initialSubscriptionName("dead-order-subscription")//处理超时订单
                        .build())
                .subscribe();


        while (true) {
            Message msg = consumer.receive();

            String orderId = (String) msg.getValue();
            boolean isPaid = checkOrderStatus(orderId);

            if (isPaid) {
                System.out.println("order " + orderId + " is paid");
                // 订单已支付，发送确认通知, 消息被成功消费
                consumer.acknowledge(msg);
            } else {
                // 订单未支付，移入重试队列
                System.out.println("order " + orderId + " is not paid");
                consumer.reconsumeLater(msg, 60, TimeUnit.SECONDS);
                // 消费者会自动订阅重试队列，60s后再次检查重试队列里的订单，若仍未支付，则消息被自动移入死信队列
            }
        }

    }

    private static boolean checkOrderStatus(String orderId) {
        // read from database
        return new Random().nextBoolean();
    }

}
